
                /*********************************************
                **                                          **
                **                                          **
                **              ca.poll.cml.C               **
                **                                          **
                **                                          **
                *********************************************/
                /*********************************************
                **                                          **
                **         Author Jean-Marc Adamo           **
                **       Copyright (C) 1997-2008 JMA        **
                *********************************************/

/*
** Asynchronous Communication
** Needs exactly 4 processes to execute.
*/

#include "arch.h"

Global_id gid1;
Coll_id cid1, cid2, cid3;

void init_gid(int, char **){
   gid1.set();
   cid1.set();
   cid2.set();
   cid3.set();
   fprintf(Sdf::trace, "node%d: gid1=%d, cid1=%d, cid2=%d, cid3=%d\n",
                Sdf::PROC_ID, gid1.get(), cid1.get(), cid2.get(), cid3.get());
   fflush(Sdf::trace);
}

int comm_rank;
int comm_size;
CommL *comm;

class SR_count{
   public:
      int cnt;
      SR_count(){
         cnt = 0;
      }
      void reset(){
         cnt = 0;
      }
      void more_SR(int i){
         cnt += i;
      }
};

#define wait_while(condition) while(condition) Sdf::poll();
#define not_complete  (c.cnt >0)

ChanA<int> *chan;

void upon_end_of_SR_r(ChanAs *chan){

   fprintf(Sdf::trace, "node%d, r>>>>>>>\n", Sdf::PROC_ID);
   Sdf::Access_type access_type;
   int tag_dv = chan->get_tag_dv(&access_type);
   fprintf(Sdf::trace, "node%d, upon_end_of_SR_r, rank=%d, capa=%d, rcvd size=%d, Tsize=%d, rps=%d, rpr=%d, tag_dv=%d, access_type=%d, tag_av=%d, tag_ev=%d, comm=%d, comm_type=%d\n", 
                Sdf::PROC_ID, 
                comm_rank, 
                chan->get_chan_capacity(), 
                chan->get_msg_size(Sdf::RCV), 
                chan->get_Tsize(),
                chan->get_remote_proc(Sdf::SND), 
                chan->get_remote_proc(Sdf::RCV),
                tag_dv,
                access_type,
                chan->get_tag_av(),
                chan->get_tag_ev(),
                chan->get_comm(),
                chan->get_comm()->get_Comm_type());

   ((SR_count *)(chan->get_hndlr_data_ptr(Sdf::RCV)))->cnt--;

   fprintf(Sdf::trace, "node%d, r<<<<<<<\n", Sdf::PROC_ID);
   fflush(Sdf::trace);
}

void upon_end_of_SR_s(ChanAs *chan){           

   fprintf(Sdf::trace, "node%d, s>>>>>>>\n", Sdf::PROC_ID);
   Sdf::Access_type access_type;
   int tag_dv = chan->get_tag_dv(&access_type);
   fprintf(Sdf::trace, "node%d, upon_end_of_SR_s, rank=%d, capa=%d, sent size=%d, Tsize=%d, rps=%d, rpr=%d, access_type=%d, tag_dv=%d, tag_av=%d, tag_ev=%d, comm=%d, comm_type=%d\n", 
                Sdf::PROC_ID, 
                comm_rank, 
                chan->get_chan_capacity(), 
                chan->get_msg_size(Sdf::SND), 
                chan->get_Tsize(),
                chan->get_remote_proc(Sdf::SND), 
                chan->get_remote_proc(Sdf::RCV),
                tag_dv,
                access_type,
                chan->get_tag_av(),
                chan->get_tag_ev(),
                chan->get_comm(),
                chan->get_comm()->get_Comm_type());

   ((SR_count *)(chan->get_hndlr_data_ptr(Sdf::SND)))->cnt--;

   fprintf(Sdf::trace, "node%d, s<<<<<<<\n", Sdf::PROC_ID);
   fflush(Sdf::trace);
}

class  Root_thread: public Thread{
   int recvd;
   public:
        Root_thread()
                : Thread((Tmf0_ptr)&Root_thread::body){}

        void body(){

            fprintf(Sdf::trace, "node%d, in Root_thread::body0\n", Sdf::PROC_ID);
            fflush(Sdf::trace);

              //last process is excluded from comm, the other processes
              //are ranked in reverse order
            int colour;
            if(Sdf::PROC_ID==Sdf::PROC_NB-1)
               colour=Sdf::UNDEFINED;
            else
               colour=1;

              //Sdf::UNDEFINED is not counted as a colour
            CommL **comm_table = Sdf::COMML_WORLD->split(cid1, 1, colour, Sdf::PROC_NB-2-Sdf::PROC_ID);
            comm = *comm_table;

              //last process is just terminated
            if(Sdf::PROC_ID==Sdf::PROC_NB-1)
               return;

            comm_rank = comm->rank();
            comm_size = comm->size();

            fprintf(Sdf::trace, "node%d, in Root_thread::body1\n", Sdf::PROC_ID);
            fflush(Sdf::trace);

            chan = new ChanA<int>(gid1, (comm_rank+1)%comm_size, 1, comm);

            fprintf(Sdf::trace, "node%d, in Root_thread::body2 rank=%d next-rank=%d, comm_size=%d, rps=%d, rpr= %d\n", 
                    Sdf::PROC_ID, comm_rank, (comm_rank+1)%comm_size, comm_size,
                    chan->get_remote_proc(Sdf::SND), chan->get_remote_proc(Sdf::RCV));
            fflush(Sdf::trace);

            SR_count c;
            chan->set(Sdf::SND, upon_end_of_SR_s, &c);
            chan->set(Sdf::RCV, upon_end_of_SR_r, &c);
            fprintf(Sdf::trace, "node%d, in Root_thread::body3\n", Sdf::PROC_ID);
            fflush(Sdf::trace);

            Sdf::barrier(cid2, comm);
                //synchronizes chan construction over comm(no longer
                //necessary starting from version 5)

            int sum_recvd = 0;
            for(int i=0; i<5; i++){
               int sent_data = Sdf::PROC_ID*10 + i;
               c.more_SR(2);
               chan->send(Sdf::STD, sent_data);
               chan->recv(recvd);
               wait_while(not_complete);
               Sdf::barrier(cid3, comm);
               //global completion over comm (actually not necessary since
               //since allreduce below cannot consume recvd before
               //it is locally available)

               int reduce_out;
                 //t_reduce is overloaded, which instance is actually used
                 //depends on the call arguments
               _Sdf<int>::reduce(recvd, reduce_out, 1, Opa<int>::SUM, cid1, comm);
               sum_recvd += reduce_out;

               fprintf(Sdf::trace, "node%d, i=%d, recvd=%d, reduce_out=%d, sum_recvd=%d\n", 
               Sdf::PROC_ID, i, recvd, reduce_out, sum_recvd);
               fflush(Sdf::trace);
            }

            Sdf::barrier(cid2, comm);
                 //this is to protect chan destruction
            delete chan;
            delete []comm_table;
            delete comm;

       }
};

MAIN(Root_thread, init_gid, Sdf::empty, Sdf::empty)
